package com.atguigu.flinksql.day13.udf;

import com.atguigu.datastream.bean.WaterSensorAvg;
import org.apache.flink.table.functions.AggregateFunction;

/**
 * ClassName: AvgFunction
 * Package: com.atguigu.flinksql.day13.udf
 * Description:
 *
 * @Author ChenJun
 * @Create 2023/4/21 19:56
 * @Version 1.0
 */
public class AvgFunction extends AggregateFunction<Double, WaterSensorAvg> {

    @Override
    public WaterSensorAvg createAccumulator() {
        return new WaterSensorAvg();
    }


    @Override
    public Double getValue(WaterSensorAvg waterSensorAvg) {
        if (waterSensorAvg.getCount() ==0L){
            return null;
        }else{
            return waterSensorAvg.getVc() / waterSensorAvg.getCount();
        }
    }


    //来一条处理一条
    public void accumulate(WaterSensorAvg acc, Double iValue) {
        acc.setVc(acc.getVc() + iValue);
        acc.setCount(acc.getCount() + 1L);
    }

    //撤回
    public void retract(WaterSensorAvg acc, Double value) {
        acc.setVc(acc.getVc() - value);
        acc.setCount(acc.getCount() - 1L);
    }

    //重置
    public void resetAccumulator(WaterSensorAvg acc) {
        acc.setVc(0.0D);
        acc.setCount(0L);
    }

}
